package org.apache.hadoop.hdfs.server.datanode;

import java.io.DataOutputStream;
import java.util.ArrayList;
import java.util.List;

/**
 * Keeps track of the progress made against each request for a block.
 * @author hadoopuser
 *
 */

public class BlockReadingProgress {


	private long requestedStartOffset;
	private long requestedEndOffset;
	private long requestedDataVolume;
	private long completedDataVolume = 0;
	
	private boolean debugMode = false;
	
	
	private long[] chunkCompletionFlags;
	private long[] targetChunkCompletionFlags;
	
	private int bytesPerChecksum;
	private long blockLength;
	private long chunksInBlock;
	private boolean finishedLatch = false;

	public BlockReadingProgress(long requestedStartingOffset, long requestedEndingOffset, long blockLength, int bytesPerChecksum) throws Exception {
		
		if (debugMode) {
			System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Creaing block Reading progress object. Start: " + requestedStartingOffset + " End: " + requestedEndingOffset + " BlockSize: " + blockLength + " Bytes per chk: " + bytesPerChecksum);
		}
		if (requestedStartingOffset <= requestedEndingOffset) {
			this.requestedEndOffset = requestedEndingOffset;
			this.requestedStartOffset = requestedStartingOffset;
		}
		else {
			if (debugMode) {
				System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Invalid offsets - start must be before end. Start" + requestedStartingOffset + " End:" + requestedEndingOffset);
			}
			throw new Exception("Invalid offsets - start must be before end");
		}
		
		if (blockLength >= 0) {
			this.blockLength = blockLength;
		}
		else {
			if (debugMode) {
				System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Invalid block length - must be > or = to 0. Block Length " + blockLength);
			}
			throw new Exception("Invalid block length - must be > or = to 0");
		}
		
		if (requestedEndingOffset > blockLength) {
			if (debugMode) {
				System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Invalid offsets - end offset greater that total block length. Block Length: " + blockLength + " ReqEndOffset: "+ requestedEndingOffset);
			}
			throw new Exception("Invalid offsets - end offset greater that total block length");
		}
		
		if (bytesPerChecksum > 0) {
			this.bytesPerChecksum = bytesPerChecksum;
		}
		else {
			if (debugMode) {
				System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Invalid bytes per checksum - must be positive number. It is: " + bytesPerChecksum);
			}

			throw new Exception("Invalid bytes per checksum - must be positive number");
		}


		
		requestedDataVolume = requestedEndingOffset - requestedStartingOffset + 1;
		
		//System.out.println("Requested Data Volume: " + requestedDataVolume);
		//System.out.println("requestedEndingOffset " + requestedEndingOffset);
		//System.out.println("requestedStartingOffset " + requestedStartingOffset);
		
		
		if (debugMode) {
			System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Finished making BlockReadingProgress object");
		}

		
	}
	

	
	public long getRequestedEndOffset() {
		return requestedEndOffset;
	}
	

	public long getRequestedStartOffset() {
		return requestedStartOffset;
	}
	
	

	/**
	 * Return true if all the data that needs to be read (i.e. all data between start and end offsets)
	 * has been read. Otherwise, false is returned.
	 * @param currentOffset
	 * @return
	 */
	public boolean amendProgress(long chunkStart, long chunkEnd) throws Exception {
		
		// Check the chunk start and chunk end values are sane.
		

		//System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Amending progress: Completed between: " + chunkStart + " and " + chunkEnd);

		
		if (chunkStart == chunkEnd) { // Nothing to do - no actual data read
			return this.allRequiredBlocksCompleted();
		}
		
		if ((chunkStart > chunkEnd) || (chunkStart % bytesPerChecksum != 0)) {
			// Start cannot be after end, and start must be on a checksumm'ed border
			Exception exception = new Exception("Start of chunk is not start of checksummed chunk or is greater than chunk end point.");
			exception.printStackTrace();
			throw exception;
		}
		
		// Now check that the end chunk is the end of the block or is at the end of a check-summed chunk. 
		if ((chunkEnd != (blockLength - 1)) && (chunkEnd + 1) % bytesPerChecksum != 0) {
			Exception exception = new Exception("End of chunk must be end of checksummed chunk or must be end of block");
			exception.printStackTrace();
			throw exception;
		}
		
		// Now update the bitwise variables to reflect the new processing that is completed.
		
		// Determine what bits need to change (as progress could range across multiple)
		
		long volumeOfDataRead = chunkEnd - chunkStart + 1;
		
		//System.out.println("Volume of data read " + volumeOfDataRead);
		
		if (chunkStart >= this.requestedStartOffset) {
			if (chunkEnd <= this.requestedEndOffset) {
				completedDataVolume = completedDataVolume + volumeOfDataRead;
				//System.out.println("In range, updating");
			}
			else {
				//System.out.println("Out of range 1");
			}
		}
		else {
			//System.out.println("Out of range 2");
		}
		
		//System.out.println("Completed data volume is now " + this.completedDataVolume);
		
		// Return whether or not processing is now complete.
		if (debugMode) {
			if (this.allRequiredBlocksCompleted()) {
				System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " All processing now completed");				
			}
			else {
				System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " BlockReading IS NOT completed");
			}


		}

		
		
		return this.allRequiredBlocksCompleted();
		

	}
	

	
	public boolean blockCompletelyRead() {
		if ((requestedStartOffset == 0) && (requestedEndOffset == blockLength) && allRequiredBlocksCompleted()) {
			return true;
		}
		else {
			return false;
		}
	}
	
	public boolean allRequiredBlocksCompleted() {
		
		
		return (requestedDataVolume == completedDataVolume);
		
	}
	
	public String toString() {
		StringBuilder builder = new StringBuilder();
		
		builder.append("Completed Volume: " + this.completedDataVolume + " Requested Volume: " + this.requestedDataVolume);
		
		return builder.toString();
	}
	
	public boolean isChunkComplete(long startOffset, long endOffset) {
		
		return (requestedDataVolume == completedDataVolume);
	}
	
	public static void main(String args[]) {
		try {
			

			
			
			
			BlockReadingProgress progress = new BlockReadingProgress(0, 511, 511, 512);
			System.out.println(progress);
			//System.out.println(progress.allRequiredBlockedCompleted());
			progress.amendProgress(0, 511);
			System.out.println(progress);
			System.out.println(progress.allRequiredBlocksCompleted());
			System.out.println("Is 0-99 complete? " + progress.isChunkComplete(0, 99));
			System.out.println("Is 100-199 complete? " + progress.isChunkComplete(100, 199));
			
			progress.amendProgress(100, 199);
			
			System.out.println("Is 100-199 complete? " + progress.isChunkComplete(100, 199));
			progress.amendProgress(400, 499);

			System.out.println(progress);
			System.out.println(progress.allRequiredBlocksCompleted());
			progress.amendProgress(200, 399);
			System.out.println(progress);
			System.out.println(progress.allRequiredBlocksCompleted());
			//BlockReadingProgress progress2 = new BlockReadingProgress(16384, 49062, 65535, 512);
			//System.out.println(progress2);
		}
		catch (Exception ex) {
			ex.printStackTrace();
		}
		
		
	}
	
}
